package selector;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.nio.ByteBuffer;
import java.nio.channels.*;
import java.util.Iterator;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @author： pang-yun
 * @date： 2021-09-10 14:05
 * 功能目的 ：用于  多线程
 */

public class MultiServer {
    public static void main(String[] args) throws IOException {
        Selector selector = Selector.open();
        ServerSocketChannel ssc = ServerSocketChannel.open();
        ssc.configureBlocking(false);
        ssc.register(selector, SelectionKey.OP_ACCEPT,null);
        ssc.bind(new InetSocketAddress(9999));

        //获取当前服务器线程
        int processors = Runtime.getRuntime().availableProcessors();
        Worker[] workers=new Worker[processors];
        for (int i=0;i<workers.length;i++){
            workers[i]=new Worker("worker-"+i);
        }
        //线程安全  可用于多线程计数
        AtomicInteger index= new AtomicInteger();

        while (true){
            selector.select();
            Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
            while (iterator.hasNext()){
                SelectionKey key = iterator.next();
                iterator.remove();
                if (key.isAcceptable()){
                    ServerSocketChannel serverSocketChannel = (ServerSocketChannel) key.channel();
                    SocketChannel socketChannel = serverSocketChannel.accept();
                    socketChannel.configureBlocking(false);

                    workers[index.getAndIncrement()%workers.length].register(socketChannel);
                }
            }
        }
    }
    static class Worker implements Runnable{
        private Selector workerSelector;
        private Thread thread;
        private String name;
        private volatile boolean start =false;
        private ConcurrentLinkedQueue<Runnable> queue =new ConcurrentLinkedQueue<>();
        public Worker(String name) {
            this.name = name;
        }

        public void register(SocketChannel sc) throws IOException {
            if (!start){
                workerSelector=Selector.open();
                thread=new Thread(this,name);
                thread.start();
                start=true;
            }
            queue.add(()->{
                try {
                    sc.register(workerSelector,SelectionKey.OP_READ,null);
                } catch (ClosedChannelException e) {
                    e.printStackTrace();
                }
            });
           workerSelector.wakeup();
        }
        @Override
        public void run() {

            while (true){
                try {
                    workerSelector.select();
                    Runnable poll = queue.poll();
                    if (poll!=null){
                        poll.run();
                    }
                    Iterator<SelectionKey> iterator = workerSelector.selectedKeys().iterator();
                    while (iterator.hasNext()){
                        SelectionKey key = iterator.next();
                        iterator.remove();

                        if (key.isReadable()){
                            SocketChannel channel = (SocketChannel) key.channel();
                            ByteBuffer byteBuffer = ByteBuffer.allocate(16);
                            try {
                                int read = channel.read(byteBuffer);
                                if (read==-1){key.cancel();}
                                byteBuffer.flip();
                                while (byteBuffer.hasRemaining()){
                                    byte b = byteBuffer.get();
                                    System.out.print((char) b);
                                }
                            } catch (IOException e) {
                                e.printStackTrace();
                                key.cancel();
                            }
                        }

                    }
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
    }
}
